Add large-payload blob auto-purge (opt-in singleton job, worker/SDK side) - #758

Large orchestration payloads are externalized to Azure Blob Storage as
`blob:v1:<container>:<blobName>` tokens. The DTS backend stores those tokens but
cannot delete the backing blobs (it has no storage credentials) — only this SDK
can. This adds an opt-in, whole-scheduler singleton durable entity +
orchestration job (mirroring src/ExportHistory) that drains payload rows the
backend has soft-deleted and deletes their blobs, then acks so the backend can
hard-delete the rows.

Design:
- PayloadStore.DeleteAsync is virtual (default throws NotSupportedException so it
  is non-breaking for existing external subclasses); BlobPayloadStore overrides
  it to decode the token and call DeleteIfExistsAsync (idempotent).
- BlobPurgeJob (TaskEntity singleton): Create is a no-op when already Active so
  racing client processes don't disturb the running job; Run starts a fixed-id
  orchestrator.
- BlobPurgeJobOrchestrator (perpetual): fetch a batch of tombstones, delete the
  blobs with capped parallelism, ack the successful deletions (failed tokens stay
  tombstoned to retry), idle on a timer when empty, ContinueAsNew periodically.
- ExecuteBlobPurgeJobOperationOrchestrator bridges client -> entity.
- Two new unary RPCs on TaskHubSidecarService: GetTombstonedPayloads /
  AckPurgedPayloads (authoritative proto follow-up: microsoft/durabletask-protobuf#76).
- LargePayloadStorageOptions gains AutoPurge (opt-in, default false) and
  PayloadPurgeBatchSize (default 500).
- Client-side BlobPurgeJobStarter (IHostedService) ensures the singleton job when
  AutoPurge is enabled, without blocking host startup. Worker always registers the
  entity/orchestrators/activities so a client-enabled job has something to run.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

…er simplification

- Drop the `Dto` suffix now that the payload records are first-class public
  types in `Microsoft.DurableTask.Client` (`TombstonedPayload`,
  `PayloadPurgeAck`).
- Collapse the magic `500` batch-size literal into a single
  `BlobPurgeConstants.DefaultBatchSize` used everywhere.
- Rename `BlobPurgeJobStatus.Stopped` -> `Pending` (still the zero value) and
  remove the dead `Failed` member (nothing ever set it; the job self-heals).
- Make the perpetual orchestrator self-heal: wrap each cycle in try/catch so a
  transient backend/entity/activity failure logs, backs off, and continues
  instead of failing the orchestration and killing the eternal loop.
- Ack poison tokens: `DeleteExternalBlobActivity` now returns a three-way
  `BlobDeleteResult` (Deleted/Discarded/Retry). Malformed tokens are discarded
  and acked so the backend can clear the stuck row instead of re-streaming it
  forever; transient failures stay tombstoned to retry.
- Replace the single-value `BlobPurgeJobCreationOptions` record with a plain
  `int` on `BlobPurgeJob.Create`.
- Guard the client fetch RPC: `GetTombstonedPayloadsAsync` throws
  `ArgumentOutOfRangeException` unless `0 < limit < 1000`.
- Simplify `BlobPurgeJobStarter` to a fixed-instance-id fire-once: drop the
  entity-active pre-check and schedule the Create bridge once with a fixed
  instance id, retrying only until the backend is reachable.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Replaces the earlier draft shape with the finalized contract and dispositions.

Wire contract (orchestrator_service.proto): drops GetTombstonedPayloads /
AckPurgedPayloads and their messages, and splices in the canonical block
verbatim -- GetLargePayloadTombstones / ReportLargePayloadPurgeResults, the
LargePayloadPurgeDisposition and LargePayloadPurgeReason enums, and
large_payload_auto_purge_enabled = 12 on GetWorkItemsRequest.

Dispositions: removes Discarded entirely. It was success-shaped and destroyed
evidence. Every attempt now returns a disposition plus a stable reason code.
v1 tokens and malformed v2 bodies are Quarantined; unknown version prefixes,
unreachable accounts, and authorization failures are Retry. Splitting the
store's three ArgumentException cases is done with a token prefix gate in the
activity rather than new exception types, which required widening
TokenPrefixV2 to internal.

Ownership marker: uploads now stamp managed_by=dts as blob metadata on both
the compressed and uncompressed write paths, and the delete path reads the
blob's properties and passes that read's ETag as an If-Match on the delete, so
check-and-delete is atomic without a lease. A blob without the marker is left
untouched and reported DELETED / BLOB_NOT_STORE_OWNED. Metadata rather than
index tags: tags are unsupported on ADLS Gen2 and would require Storage Blob
Data Owner.

Orchestration: the batch is now always reported, since the backend owns retry
scheduling and needs to hear about failures to defer a row. Backoff is
preserved for the all-retry case so a storage outage cannot become a tight
refetch loop. Progress counts only Deleted rows; counting quarantined rows as
purged would reintroduce the success-shaped reporting this change removes.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Pre-review polish only; no behaviour change.

CS8603 in DeleteExternalBlobActivity.SanitizeErrorCode fired on netstandard2.0
only, where string.IsNullOrEmpty carries no [NotNullWhen(false)] annotation so
flow analysis cannot prove the else branch non-null. Replaced with a constant
pattern, which the compiler analyses itself and so behaves identically on every
target framework.

CA1001 in BlobPurgeJobStarter: the type owned a CancellationTokenSource but was
not disposable. The source is deliberately disposed in Dispose rather than in
StopAsync -- StopAsync stops waiting as soon as the host shutdown token fires,
so the ensure task may still hold the token, and disposing there would fault
that still-running task with ObjectDisposedException. The container disposes
singletons after every StopAsync has returned, which is the safe point.

SA1600: documented the BlobPurgeJobStarter constructor.

CA1873 on BlobPurgeJobOrchestrator is deliberately left as-is: it occurs 77
times across this repo, so unguarded log calls are the established local
convention and changing one new file would make it inconsistent with its
neighbours for no benefit.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

The managed-to-proto enum casts in ReportLargePayloadPurgeResultsAsync are safe
only because no enum crosses the wire inbound on this feature: the SDK casts
values it defined itself, so it can never receive an unknown value and silently
reinterpret it. That invariant held by the shape of the contract rather than by
construction, and nothing in the code stated it.

Adding an enum to an inbound type would create exactly the hazard the casts
avoid today -- a newer backend sending a value this SDK does not know, mapped by
raw numeric cast onto a valid-but-wrong member -- and it would compile silently,
mis-dispositioning rows in production rather than failing a build.

Asserted by reflection over the inbound types in the existing parity test file,
which already exists to protect this mapping. The casts are unchanged: they are
correct today, and an explicit switch would add branching for a hazard that does
not currently exist on this path.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

CA1859: CreateOwnershipMetadata returned IDictionary<string, string> while
always constructing a Dictionary. The method is private static with two call
sites, both assigning to BlobOpenWriteOptions.Metadata (IDictionary<string,
string>), so returning the concrete type is implicitly compatible and lets the
JIT devirtualize the indexer.

SA1117: split the ArgumentOutOfRangeException arguments one per line.

CA1873 in BlobPurgeJobOrchestrator is left deliberately: unguarded logging is
the established convention here (77 occurrences solution-wide), so guarding one
new file would make it inconsistent with its neighbours.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: b69ecb19-b596-4e46-bb44-12ce571ec31f

The backend never branches on reason - it acts on disposition alone - so
reason is diagnostics only, and its granularity should match the number of
distinct operator responses rather than the number of distinct causes.
storageErrorCode already carries the specific storage status, so four of the
old values encoded the same fact twice.

TRANSIENT_STORAGE_FAILURE, STORAGE_ACCOUNT_UNREACHABLE and
STORAGE_AUTHORIZATION_FAILED collapse into STORAGE_FAILURE. MALFORMED_TOKEN,
INVALID_STORAGE_REQUEST, LEGACY_V1_TOKEN and UNSUPPORTED_TOKEN_VERSION
collapse into TOKEN_NOT_PURGEABLE. STORE_CANNOT_DELETE stays separate because
storage is never contacted, so its storageErrorCode is empty.

Dispositions are unchanged on every branch. TOKEN_NOT_PURGEABLE now spans both
Quarantined and Retry - an unsupported version prefix stays Retry because an
SDK upgrade resolves it - so each branch continues to state its disposition
explicitly rather than deriving it from the reason, which would strand those
rows permanently.

Proto verified byte-identical to contract-canonical.proto across all three
repos by the marker-based harness.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: b69ecb19-b596-4e46-bb44-12ce571ec31f

Comment-only. No enum value, field number, disposition, or code path
changes; a non-comment-line filter over the full diff of both files
returns empty.

Block 3 of orchestrator_service.proto is re-spliced verbatim from the
canonical contract. The "--- Reported with DELETED / RETRY /
QUARANTINED ---" banners are removed: they asserted a 1:1
reason-to-disposition mapping that does not hold, because
TOKEN_NOT_PURGEABLE is deliberately reported with two dispositions
(QUARANTINED for a v1/malformed/invalid-request token, RETRY for an
unrecognized version prefix that a newer worker can read). The
disposition now lives in each value's own comment, and the enum header
states outright that reason and disposition are orthogonal and that no
mapping between them may be asserted.

LargePayloadPurgeReason XML docs are rewritten to mirror the new
canonical prose, so the managed surface no longer describes the
superseded grouping.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: b69ecb19-b596-4e46-bb44-12ce571ec31f

Both fields were verified write-only across the whole system: the backend
persists them and nothing reads either one - no SELECT, no WHERE, no API,
no alert. Disposition alone drives backend behavior, so the enum was a
lossy copy of information that already lives in worker telemetry.

Deleting reason also deletes the hazard it created. TOKEN_NOT_PURGEABLE
spanned both Retry and Quarantined, so any consumer deriving disposition
from reason would have silently stranded rows an SDK upgrade would fix.

All 12 activity branches keep their disposition unchanged. Verified branch
by branch against the pre-change file rather than by inspection: the
unknown-prefix branch stays Retry, and the v1-prefix, malformed-token, and
HTTP 400 branches stay Quarantined.

Failure detail is now logged rather than sent over the wire. Design section 7
forbids logging raw exceptions, so each classification site logs a bounded
cause literal instead. Logging at the classification site rather than at the
call site restores the full 11-way granularity the 7-value enum had lost:
401/403, 5xx, and an unreachable account are now distinct in telemetry where
the enum had collapsed all three into STORAGE_FAILURE.

Two deliberate deviations from the request, both reported:

1. LargePayloadPurgeEnumParityTests.cs is kept, not deleted. It always
   tested two enums, and the disposition cast in GrpcDurableTaskClient
   survives - it is the cast that decides whether a row is deleted or
   quarantined. Only the reason parity fact is removed.

2. This is not the last outbound enum cast. LargePayloadPurgeResult still
   carries Disposition, so the no-inbound-enum invariant test still guards
   a live cast and is not vacuous.

CA1873 at BlobPurgeJobOrchestrator.cs:66 remains deliberately unfixed:
unguarded logging is the established convention here, 77 times solution-wide.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
Copilot-Session: b69ecb19-b596-4e46-bb44-12ce571ec31f